# ray.init(address="auto", ignore_reinit_error=True)
from common.variables import TimeFrameType
from copy import deepcopy
from datetime import datetime, time as dtime, timedelta
from functools import wraps
import numpy as np
import os
import pandas as pd
import random
import ray
import streamlit as st
import time
from trade.data.table_edit import get_table_edit_obj
from trade.ray_task_control import get_task_control_obj, task_doc_list


remote_table_edit = get_table_edit_obj()
data_ttl = 600

if "use_st_cache" in os.environ and os.environ["use_st_cache"]:
    stcache = st.cache_data(ttl=data_ttl)
else:
    stcache = lambda x: x


def mock_cost_time(seconds):
    def mock_func(func):
        @wraps(func)
        def deco_func(*args, **kwargs):
            print(f"{func.__name__}, 模拟运行一段时间...{seconds}秒, {args}, {kwargs}")
            time.sleep(seconds)
            res = func(*args, **kwargs)
            return res

        return deco_func

    return mock_func


@stcache
@mock_cost_time(2)
def get_account_info(exchange, account):
    return {
        "名称": exchange,
        "账户": account,
        "总资金": 1000,
    }


def mock_get_kline_data(symbol, start, end, timeframe, adjust):
    dates = pd.date_range(end - timedelta(days=30), end, freq="D")
    return pd.DataFrame(
        {
            "open": np.random.rand(len(dates)) * 30000,
            "high": np.random.rand(len(dates)) * 31000,
            "low": np.random.rand(len(dates)) * 29000,
            "close": np.random.rand(len(dates)) * 30000,
        },
        index=dates,
    )


@stcache
def get_kline_data(symbol, start, end, timeframe, adjust):
    start = int(datetime.combine(start, dtime.min).timestamp() * 1000)
    end = int(datetime.combine(end, dtime.max).timestamp() * 1000)
    results = ray.get(
        remote_table_edit.get_factor_from_table_by_pivot.remote(
            timeframe, start, end, symbol, ["open", "high", "low", "close", "volume"]
        )
    )
    results.index = pd.to_datetime(results.index, unit="ms")
    return results


@stcache
@mock_cost_time(1)
def mock_get_factor_rank(symbols, factors, start, end, delay: int = 5):
    return pd.DataFrame(
        {
            "symbol": symbols,
            "factor": [factors[0]] * len(symbols),
            "correlation": np.random.rand(len(symbols)),
        }
    )


@stcache
def get_factor_corr(timeframe, symbols, factors, start, end, delay: int = 5):
    start = int(datetime.combine(start, dtime.min).timestamp() * 1000)
    end = int(datetime.combine(end, dtime.max).timestamp() * 1000)
    return ray.get(
        remote_table_edit.get_factor_corr.remote(
            timeframe, start, end, symbols, factors, delay
        )
    )


@mock_cost_time(2)
def operate_position(*args, **kwargs):
    return {
        "status": "success",
        "message": f"{kwargs.get('action')} position for {kwargs.get('symbol')} executed.",
        "data": {"symbol": kwargs.get("symbol"), "quantity": kwargs.get("quantity")},
        "timestamp": time.time(),
    }


@mock_cost_time(2)
def operate_order(*args, **kwargs):
    return {
        "status": "success",
        "message": f"Order {kwargs.get('order_id')} {kwargs.get('action')} request sent.",
        "data": {"order_id": kwargs.get("order_id")},
        "timestamp": time.time(),
    }


def submit_order(**kwargs):
    """
    模拟提交下单操作。
    参数应包括：exchange, account_id, symbol, order_type, side, price, quantity
    """
    print("[submit_order] called with:", kwargs)
    time.sleep(random.uniform(1.0, 2.0))

    fake_order_id = f"ORD{random.randint(10000, 99999)}"

    return {
        "status": "success",
        "message": f"Order submitted: {kwargs.get('side')} {kwargs.get('quantity')} {kwargs.get('symbol')} at {kwargs.get('price')}",
        "data": {"order_id": fake_order_id},
        "timestamp": time.time(),
    }


@stcache
@mock_cost_time(1)
def get_exchange():
    return ["Binance", "OKX"]


@stcache
@mock_cost_time(1.1)
def get_account(*args):
    return ["Alice", "Bob"]


def get_balance_account(*args):
    return []


@stcache
@mock_cost_time(1.3)
def get_positions(*args, **kwargs):
    return [
        {
            "symbol": "BTC/USDT",
            "position_side": "long",
            "entry_price": 28500.0,
            "mark_price": 28720.5,
            "unrealized_pnl": 220.5,
            "leverage": 10,
            "position_size": 0.5,
            "margin_type": "isolated",
            "timestamp": "2025-05-04T20:30:00Z",
        },
        {
            "symbol": "ETH/USDT",
            "position_side": "short",
            "entry_price": 1800.0,
            "mark_price": 1785.0,
            "unrealized_pnl": 18.0,
            "leverage": 5,
            "position_size": 2.0,
            "margin_type": "cross",
            "timestamp": "2025-05-04T20:31:00Z",
        },
    ]


@stcache
@mock_cost_time(1.7)
def get_orders(*args, **kwargs):
    orders_info = [
        {
            "order_id": "ord001",
            "symbol": "BTC/USDT",
            "side": "buy",
            "type": "limit",
            "price": 28400.0,
            "amount": 0.1,
            "status": "open",
            "filled": 0.0,
            "remaining": 0.1,
            "timestamp": "2025-05-04T20:32:00Z",
        },
        {
            "order_id": "ord002",
            "symbol": "ETH/USDT",
            "side": "sell",
            "type": "market",
            "price": None,  # 市价单无价格
            "amount": 1.5,
            "status": "closed",
            "filled": 1.5,
            "remaining": 0.0,
            "timestamp": "2025-05-04T20:33:00Z",
        },
        {
            "order_id": "ord003",
            "symbol": "SOL/USDT",
            "side": "buy",
            "type": "limit",
            "price": 25.3,
            "amount": 10,
            "status": "canceled",
            "filled": 0.0,
            "remaining": 10.0,
            "timestamp": "2025-05-04T20:34:00Z",
        },
    ]
    return orders_info


@stcache
@mock_cost_time(2)
def get_trade_histories(*args, **kwargs):
    trades_info = [
        {
            "trade_id": "tx001",
            "order_id": "ord002",
            "symbol": "ETH/USDT",
            "side": "sell",
            "price": 1799.5,
            "amount": 1.5,
            "fee": {"currency": "USDT", "cost": 1.35},
            "timestamp": "2025-05-04T20:36:10Z",
        },
        {
            "trade_id": "tx002",
            "order_id": "ord004",
            "symbol": "BTC/USDT",
            "side": "buy",
            "price": 28600.0,
            "amount": 0.05,
            "fee": {"currency": "USDT", "cost": 0.5},
            "timestamp": "2025-05-04T20:37:20Z",
        },
        {
            "trade_id": "tx003",
            "order_id": "ord005",
            "symbol": "SOL/USDT",
            "side": "buy",
            "price": 25.1,
            "amount": 5,
            "fee": {"currency": "USDT", "cost": 0.25},
            "timestamp": "2025-05-04T20:38:05Z",
        },
    ]
    return trades_info


@stcache
@mock_cost_time(1.5)
def mock_get_factor_list():
    return ["factor_aaa", "factor_bbb", "factor_ccc"]


@stcache
def get_factor_list(timeframe: TimeFrameType):
    return ray.get(remote_table_edit.get_distinct_factors.remote(timeframe))


@stcache
@mock_cost_time(1)
def mock_get_symbol_list():
    return ["BTC/USDT", "ETH/USDT", "AXR/USDC"]


@stcache
def get_symbol_list(timeframe: TimeFrameType):
    return ray.get(remote_table_edit.get_distinct_symbols.remote(timeframe))


@stcache
@mock_cost_time(1)
def mock_get_factor_values(symbol_list, factor_list, start, end, timeframe):
    """
    This function takes the following inputs:
    - `symbol_list`: A list of trading symbols (e.g., ["BTC/USDT", "ETH/USDT"]).
    - `factor_list`: A list of factor names (e.g., ["Momentum", "Volatility"]).
    - `start`: The start date for the data (datetime or string).
    - `end`: The end date for the data (datetime or string).
    - `timeframe`: The timeframe for the data (e.g., "1d", "1h").

    Returns:
    - A pandas DataFrame with:
      - Index: Time (datetime) within the specified range.
      - Columns: Each column is a tuple `(symbol, factor_name)` representing the factor value for the symbol.
    - The returned values are randomly generated as a placeholder.
    """
    dates = pd.date_range(start, end, freq=timeframe)
    data = {
        (symbol, factor): np.random.randn(len(dates))
        for symbol in symbol_list
        for factor in factor_list
    }
    return data


@stcache
def get_factor_values(
    symbol_list, factor_list, start: datetime, end: datetime, timeframe
):
    start = int(datetime.combine(start, dtime.min).timestamp() * 1000)
    end = int(datetime.combine(end, dtime.max).timestamp() * 1000)
    rs = ray.get(
        remote_table_edit.get_factor_from_table.remote(
            timeframe, start, end, symbol_list, factor_list
        )
    )
    if rs is None or len(rs) <= 0:
        return None
    rs.set_index("xdatetime", inplace=True)
    rs.sort_values("xdatetime", inplace=True)
    rs.index = pd.to_datetime(rs.index, unit="ms")
    results = {key: value["value"] for key, value in rs.groupby(["symbol", "fname"])}
    return results


@stcache
def get_kline_data_by_pivot(symbol, start, end, timeframe):
    start = int(datetime.combine(start, dtime.min).timestamp() * 1000)
    end = int(datetime.combine(end, dtime.max).timestamp() * 1000)
    results = ray.get(
        remote_table_edit.get_factor_from_table_by_pivot.remote(
            timeframe, start, end, symbol, ["open", "high", "low", "close", "volume"]
        )
    )
    if results is None or len(results) <= 0:
        return None
    results.index = pd.to_datetime(results.index, unit="ms")
    results.sort_index(inplace=True)
    return results


@stcache
@mock_cost_time(1)
def get_strategy_list():
    """
    返回多种策略的信息列表，返回值中每个元素是一个字典，包括了策略名，策略参数描述的dict，策略代码所在的文件位置，策略添加时间
    :return:
    """
    return [
        {
            "name": "strategy_aaa",
            "params": [
                {
                    "name": "param1",
                    "type": "int",
                    "description": "Parameter 1 description",
                },
                {
                    "name": "param2",
                    "type": "float",
                    "description": "Parameter 2 description",
                },
            ],
            "module_path": "strategies.strategy_aaa",
            "added_time": "2025-01-01T10:00:00Z",
        },
        {
            "name": "strategy_bbb",
            "params": [
                {
                    "name": "param1",
                    "type": "int",
                    "description": "Parameter 1 description",
                },
                {
                    "name": "param2",
                    "type": "str",
                    "description": "Parameter 2 description",
                },
                {
                    "name": "param3",
                    "type": "bool",
                    "description": "Parameter 3 description",
                },
            ],
            "module_path": "strategies.strategy_bbb",
            "added_time": "2025-01-05T12:30:00Z",
        },
        {
            "name": "strategy_ccc",
            "params": [
                {
                    "name": "param1",
                    "type": "float",
                    "description": "Parameter 1 description",
                }
            ],
            "module_path": "strategies.strategy_ccc",
            "added_time": "2025-01-10T14:45:00Z",
        },
    ]


def add_task(task_name: str, title: str, params: dict):
    obj = get_task_control_obj()
    rs = ray.get(obj.add_task.remote(task_name, title, **params))
    return rs


def cancel_task(task_id: str):
    obj = get_task_control_obj()
    rs = ray.get(obj.cancel_task.remote(task_id))
    return rs


def get_can_do_tasks():
    return deepcopy(task_doc_list)


@stcache
def read_exp_info():
    obj = get_task_control_obj()
    results = ray.get(obj.read_exp_info.remote())
    return results


@stcache
def read_exp_result(task_id: str):
    obj = get_task_control_obj()
    results = ray.get(obj.read_exp_result.remote(task_id))
    return results
